fix(replication): drain priority sync queue and recover from routing-event lag#165
Open
mickvandijke wants to merge 27 commits into
Open
fix(replication): drain priority sync queue and recover from routing-event lag#165mickvandijke wants to merge 27 commits into
mickvandijke wants to merge 27 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR improves replication convergence under churn by (a) ensuring neighbor sync drains queued priority peers without waiting for periodic ticks and (b) recovering from lagged DHT routing-table broadcast events. It also expands verification/fetch pipeline behavior (batching caps, retry/defer semantics, paid-list edge-voter quorum handling) and adds an e2e scenario covering paid-list-authorized repair below storage quorum.
Changes:
- Neighbor-sync loop now drains the priority queue back-to-back and resyncs close peers on
RecvError::Lagged. - Verification/fetch pipeline enhancements: bounded verification batches, fetch→verification retry metadata, and deferred re-verification scheduling.
- Paid-list quorum evaluation updated with “edge voter” handling; adds a deterministic paid-list repair e2e scenario and bumps crate version.
Reviewed changes
Copilot reviewed 11 out of 12 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/poc_d1_bounded_queues.rs | Updates test VerificationEntry construction for new timing fields. |
| tests/poc_bootstrap_stall.rs | Updates test VerificationEntry construction for new timing fields. |
| tests/e2e/replication.rs | Adds deterministic e2e scenario for paid-list-majority repair under low storage quorum. |
| src/replication/types.rs | Adds next_verify_at, fetch retry metadata, and NeighborSyncState::has_priority_peers + test. |
| src/replication/scheduling.rs | Adds deferred pending scheduling, fetch retry→verification requeue, and returns evicted keys. |
| src/replication/quorum.rs | Adds edge-aware paid-list vote summary and splits verification requests into capped batches. |
| src/replication/mod.rs | Implements lag recovery, neighbor-sync drain-before-park, verification request caps, and retry/defer integration. |
| src/replication/config.rs | Introduces PAID_LIST_FLEX_EDGE_COUNT and MAX_VERIFICATION_KEYS_PER_REQUEST. |
| src/replication/bootstrap.rs | Updates test VerificationEntry construction for new timing fields. |
| src/replication/admission.rs | Adjusts cross-set dedup/admission so paid hints can survive replica rejection under churn. |
| Cargo.toml | Bumps version to 0.14.3. |
| Cargo.lock | Updates locked crate version to 0.14.3. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
2543
to
2547
| results.push(protocol::KeyVerificationResult { | ||
| key: *key, | ||
| present, | ||
| present: cached.present.unwrap_or(false), | ||
| paid, | ||
| }); |
| } | ||
| continue; | ||
| } | ||
| Err(RecvError::Closed) => continue, |
Comment on lines
+495
to
+497
| let handles = | ||
| spawn_verification_batch_tasks(targets, p2p_node, config.verification_request_timeout); | ||
| collect_verification_batch_results(handles, targets, &mut evidence).await; |
Comment on lines
+43
to
+47
| /// Number of furthest paid-list close-group peers treated as churny edge | ||
| /// voters. | ||
| /// | ||
| /// Once the paid-list close group reaches [`PAID_LIST_CLOSE_GROUP_SIZE`], edge | ||
| /// peers are queried, but a negative edge paid-list response does not count |
mickvandijke
force-pushed
the
fix/neighbor-sync-drain-window
branch
from
1a6fe63 to
fc8d724
Compare
Comment on lines
+1284
to
+1288
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; |
Comment on lines
+2837
to
2841
| if cached.present.is_none() && paid.is_none() { | ||
| continue; | ||
| } | ||
|
|
||
| results.push(protocol::KeyVerificationResult { |
Comment on lines
18
to
21
| pub mod audit; | ||
| pub mod audit_coordinator; | ||
| pub(crate) mod audit_metrics; | ||
| pub mod bootstrap; |
Comment on lines
+132
to
+134
| fn fresh_offer_key_lock_index(key: &XorName) -> usize { | ||
| usize::from(key[0]) % FRESH_OFFER_KEY_LOCK_SHARDS | ||
| } |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
mickvandijke
force-pushed
the
fix/neighbor-sync-drain-window
branch
from
74c990e to
b8d490d
Compare
Comment on lines
+631
to
+633
| self.insert_pending_unchecked(key, verification); | ||
| self.release_retry_slot(&sender); | ||
| true |
Comment on lines
+53
to
+54
| #[cfg(feature = "logging")] | ||
| impl AuditType { |
Comment on lines
+77
to
+78
| #[cfg(feature = "logging")] | ||
| impl AuditResponderClass { |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
Comment on lines
+204
to
+209
| if let Some(retry) = &mut candidate.retry_verification { | ||
| retry | ||
| .replica_hint_sources | ||
| .extend(entry.replica_hint_sources); | ||
| retry.hint_sources.extend(entry.hint_sources); | ||
| } |
Comment on lines
+222
to
+227
| if let Some(retry) = &mut in_flight.retry_verification { | ||
| retry | ||
| .replica_hint_sources | ||
| .extend(entry.replica_hint_sources); | ||
| retry.hint_sources.extend(entry.hint_sources); | ||
| } |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
Comment on lines
+110
to
+114
| enum PruneAuditChallengeResult { | ||
| Response(Box<ReplicationMessage>), | ||
| NoResponse(AuditFailureClass), | ||
| MalformedResponse, | ||
| } |
Comment on lines
+1183
to
+1187
| PruneAuditChallengeResult::Response(decoded) => *decoded, | ||
| PruneAuditChallengeResult::NoResponse(class) => { | ||
| // No response means an immediate audit failure, but keep the local | ||
| // class split so timeout metrics are not polluted by pre-delivery | ||
| // failures. |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
Comment on lines
+321
to
+326
| if let Some(keys) = self.pending_keys_by_source.get_mut(source) { | ||
| keys.remove(key); | ||
| if keys.is_empty() { | ||
| self.pending_keys_by_source.remove(source); | ||
| } | ||
| } |
Comment on lines
+115
to
+121
| let Some(entry) = targets.get_mut(&peer) else { | ||
| return; | ||
| }; | ||
| entry.references = entry.references.saturating_sub(1); | ||
| if entry.references == 0 { | ||
| targets.remove(&peer); | ||
| } |
Comment on lines
+53
to
+54
| #[cfg(feature = "logging")] | ||
| impl AuditType { |
Comment on lines
+77
to
+78
| #[cfg(feature = "logging")] | ||
| impl AuditResponderClass { |
Comment on lines
+153
to
157
| // Cross-set precedence: the replica pass already ruled on this key, | ||
| // under the same gate this pass would apply. Re-asking cannot change | ||
| // the answer, and re-recording it would double-count the rejection. | ||
| if seen_replica.contains(&key) { | ||
| continue; |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
SemVer: patch
…event lag Neighbor-sync paid-list hint propagation could stall for tens of minutes on the furthest close-group members during correlated topology change (partition heal, mass join). Two root causes: - The neighbor-sync loop parked on the periodic 10-20 min tick after every single round, and `sync_trigger` is a coalescing `Notify` that collapses a burst of entrant wakeups into one. A churn burst that queued many priority peers therefore drained only one batch (<=4) promptly; the rest waited for subsequent ticks. Park only when `priority_order` is empty and otherwise run rounds back-to-back, draining the durable queue at round-trip speed. The drain terminates because `select_next_sync_peer` pops each priority peer unconditionally and `new_cycle` only refills under `is_cycle_complete()`. - The DHT event handler discarded broadcast `RecvError::Lagged`, silently dropping `KClosestPeersChanged` events under load -- exactly when churn is heaviest -- so missed entrants were never queued. On lag, resynchronize from ground truth: snapshot the current close-peer set, queue every member for priority sync, and fire the trigger. Add `NeighborSyncState::has_priority_peers` and a regression test covering the loop's drain/termination contract. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
SemVer: patch Consume dequeued fetch candidates through reservation-aware queue APIs and requeue no-source retry candidates for verification. Centralize pending_verify insertion bookkeeping so per-sender counters stay in lockstep.
SemVer: patch
A closed Tokio broadcast receiver is immediately ready with RecvError::Closed on every recv(), and both the P2P and DHT event branches responded by continuing the select! loop — spinning a core (and, on the P2P branch, flooding logs) until shutdown. A closed broadcast channel can never yield again, so treat Closed as terminal: handle_replication_event_recv_error now returns ControlFlow and both branches break the loop, which also drops replication_tx and cascades a clean shutdown of the serial handler. Also hoist mid-file imports flagged by the rust-conventions hook: the saorsa_pqc sig import moves to the top block and the tests module switches to `use super::*;` with module-qualified paths. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The broadcast-lag recovery path requeued the current close-peer snapshot but never called retain_sync_peers, unlike the normal KClosestPeersChanged path. Peers that left the close set during the lost event window stayed in priority_order ahead of genuine entrants, each burning a request timeout before current peers could sync. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Once all four worker permits were held, dispatch_fresh_offer handled the next offer inline — an on-chain payment verification and a multi-MiB LMDB write — on the serial non-audit loop. Stalling that loop backs up the 256-slot inbound queue, at which point the P2P event loop starts handling messages inline too, and the broadcast receiver lags and drops replication messages wholesale. The per-key shard locks made that easy to reach. The index was key[0] % 64, but a node only receives fresh offers for keys close to its own ID, so the accepted key set shares a long prefix and nearly every offer landed on one shard. Unrelated keys serialized behind a single mutex, keeping the four workers occupied and making the inline path the steady state rather than the exception. The ordering those locks preserved has no meaning here: the key is the content address of the data, so same key implies same bytes. Dispatch now only claims the key, takes an admission permit, and spawns. The worker permit is awaited inside the task, so handle_fresh_offer has a single caller and no inline path exists to fall back to. A new admission semaphore bounds outstanding offers — each holds its payload until it completes — at 16, a 64 MiB ceiling. The shard locks are replaced by an exact per-key in-flight set behind an RAII guard, so unrelated keys never contend and concurrent duplicates collapse onto the first claimant rather than repeating its verification. Behaviour change: past the admission bound an offer is refused rather than queued or handled inline. A refused offer is not stored and is therefore penalized as absent by the delayed possession check, the same as any other declined replica. The previous behaviour was not penalty-free either — a stalled loop drops offers through broadcast lag, with the same result and less predictability. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A replica hint is a possession claim made by the sending peer. The receiver treated it as authorization: `HintPipeline::Replica` skipped the `is_responsible(storage_admission_width)` check that `PaidOnly` keys had to pass, so whoever labelled the hint decided what we store. Two messages were enough to exploit it. A paid hint for key K admits at `paid_list_close_group_size` (20) and lands in `pending_verify` as `PaidOnly`. A second message re-advertising K as a replica hint hits the `already_pending` fast path in `admit_hints`, which skips admission checks for keys already queued, and reaches `add_pending_verify` as `Replica` — escalating the live entry. Both fetch paths then honoured the tag and downloaded without ever asking whether this node was in the top-9 for K. Nodes ranked 10-20 could be conscripted into fetching and storing any key an attacker could name; pruning reclaims it, but the bandwidth and disk pressure are spent, and honest routing skew triggers the same path. Derive `pipeline` from `replica_hint_sources` instead of storing it. The tag *is* "did any peer claim to hold this", so a stored copy could only drift from its own definition — `remove_hint_source` already recomputed it on peer departure. Deriving deletes the escalation and both demotion sites. Then gate downloads where the decision is actually made, on live routing state, in both places the pipelines diverged: the local paid-list fast path and the post-verification path. `pipeline` keeps its real jobs — selecting fetch sources and scoping sole-source trust penalties — and loses the one it should never have had. This narrows replica repair from the sender's view to our own: a node with a transiently skewed routing table now declines to repair a key it ranks outside the top-9. That matches pruning, which already evicts at the same width, so the fetch would have been undone anyway. Fresh PUTs keep their wider accept window (see `handle_fresh_replication_offer`), where rejecting risks losing data a client is writing now. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Admission asked two different questions depending on which list the sender put a key on: replica hints were gated at storage_admission_width (9), paid hints at paid_list_close_group_size (20). The sender chose which gate applied to its own hint. Ask one question instead. Admission decides only relevance -- should we learn this key exists and is paid for -- which is the 20-wide paid close group, since that is the width across which nodes track payment validity. Storage responsibility stays where the previous commit put it: at the point of download, against live routing state. Mislabelling a hint now gains an attacker nothing, because both labels reach the same gate. The hint set only records whether the sender claimed possession, which makes it a candidate fetch source. The set of admissible keys is unchanged -- a paid hint for any key a replica hint could carry was already admissible at 20 -- so this widens no attack surface. It does recover information the old gate discarded: a replica hint for a key we rank 10-20 for was rejected outright, even though that band exists precisely to track payment validity. The two-gate dance this removes (rejected_replica rescued by admitted_paid) existed only because the gates disagreed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… heap A hint for a key already in the fetch queue only needs its advertiser merged into that candidate's source set -- a field the heap never orders on. The old path took the whole heap, scanned it linearly for the key, and re-heapified, costing O(n) per duplicate. Under source aggregation a batch of m duplicates ran O(m*n) while holding the global queue write lock, which a neighbor could trigger deliberately by re-hinting keys it knows are queued. Split FetchCandidate into FetchOrder (key + distance, the only fields the Ord impl reads) held in the heap, and FetchPayload (sources + retry metadata) held in a key-indexed map that also serves as the membership index. Merging a source is now an O(1) map lookup that never touches the heap; the ordering invariant is immutable-by-construction rather than restored by a rebuild. The departed-peer path edits payloads in place and rebuilds only when a peer actually orphaned a candidate. Measured per-key merge cost goes from 302us at a 50k-deep queue to a flat ~400ns independent of depth. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
ReplicationEngine::shutdown() promises that no background work still holds the LMDB environment when it returns, but several paths race a select! on the shutdown token against futures awaiting a spawn_blocking LMDB transaction (fetch storage.put, prune storage.delete / paid_list.remove_batch, verification paid_list.insert). Dropping the losing future does not cancel spawn_blocking: the closure keeps running with a cloned Env, so reopening the same environment after shutdown was undefined behavior. Track the blocking tasks at the storage layer instead of shielding call sites: LmdbStorage and PaidList each route every spawn_blocking through their own TaskTracker and expose wait_idle(); shutdown() awaits both after its existing detached-task drain. Per-fetch tasks are additionally spawned on the detached tracker so a dropped in_flight set can no longer leak Arc<LmdbStorage> past shutdown. Cancellation semantics are unchanged — network I/O still aborts promptly; only the bounded in-flight transaction is awaited. Regression coverage: storage- and paid-list-level wait_idle tests park a write inside its blocking closure and drop the awaiter; a new poc_shutdown_lmdb_drain suite proves shutdown() blocks until the detached write commits and both environments reopen cleanly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ll bootstrap A capacity-rejection record for source P is only cleared by P's next clean admission cycle or by P's PeerRemoved cleanup. The note sites and the PeerRemoved handler run on different tokio tasks with await points between the last "P is live" observation and the insert, so a removal processed inside that window makes its clear a no-op and the subsequent note records an entry no future event can retire. check_bootstrap_drained then returns false forever: audits stay disabled (Invariant 19) and the node advertises bootstrapping:true indefinitely, drawing network-wide bootstrap-claim trust penalties — reachable deliberately by overflowing pending_verify during a victim's bootstrap and disconnecting. Track each source's most recent rejection time instead of a bare set and expire records older than CAPACITY_REJECTED_MAX_AGE (three neighbor-sync intervals at the slowest cadence plus slack — a live source re-hints every cycle, so silence that long means re-delivery was abandoned). Expiry forfeits the departed source's owed keys, consistent with update_bootstrap_after_peer_removed; post-bootstrap neighbor sync and audit/repair recover them. Run expiry plus a drain re-check from the verification worker tick, ahead of the pending_peer_requests early-returns: pending requests legitimately block the drain but must not block expiry. This also closes an adjacent liveness gap — every drain check was event-driven and a clean-cycle clear never re-checked, so a quiet node could satisfy the drain condition with nothing left to observe it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…load Storage responsibility was checked once, at verification-completion time, and never again before the chunk was downloaded and stored. The fetch queue holds up to 131,072 entries and dequeues nearest-first, so a far candidate can wait unboundedly long while closer keys jump ahead — topology churn in that window could move this node out of the storage-admission group for a promoted key, yet the fetch worker still downloaded and stored it: wasted bandwidth, a wasted disk write, and fetch→store→prune churn once the prune pass evicted the record again. The per-source retry path reused the same stale decision for every fallback source. types.rs and admission.rs both documented the intended contract — responsibility is decided against live routing state at the point of download — and the code violated both. Recheck responsibility per fetch attempt inside execute_single_fetch, where no queue lock is held: once at the top, before spending bandwidth (every per-source retry re-enters here, so retries are covered), and once more before storage.put, so bytes that arrive after responsibility lapsed mid-round-trip are not written. Edge flapping is dampened by the margin storage_admission_width already adds over close_group_size. A lapsed attempt resolves as NoLongerResponsible, which shares Stored's terminal path — complete_fetch releases the verification retry-slot reservation and the worker's terminal handling shrinks the bootstrap pending set, so bootstrap drain cannot stall on a declined key. No trust event is reported: the source did nothing wrong. The verification-time check stays as a cheap pre-filter that keeps never-responsible keys out of the fetch queue; the per-attempt check is the authoritative gate. Event-driven purging of the fetch queue on routing changes was considered and rejected — O(queue) scans per churn event, racy, and strictly more complex than the lazy per-attempt recheck. The worker's result handling is extracted into apply_fetch_result so the per-variant queue transitions are unit-tested directly (terminal exit, retry-slot release, no verification requeue, preserved failure paths). A new e2e drives a live 12-node network: a test-only seam enqueues a fetch candidate for a key the target is not responsible for — the exact state a stale promotion leaves behind — and asserts the chunk is never stored and the key exits the pipeline terminally, with an in-responsibility positive control through the same seam and holder. Live topology cannot be shifted deterministically inside the promotion→dequeue window (random peer IDs, no routing-table injection API, millisecond timing), hence the seam. With the rechecks disabled the e2e fails by storing the stale chunk, confirming it discriminates. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add ADR-0005 capturing the design decisions behind the replication repair hardening work: responsibility decided at download against live routing state, fresh offers off the serial loop, detached async/LMDB lifecycle tracking, terminal closed streams, eager neighbor-sync drain, TTL'd bootstrap accounting, source-aware bounded verification, retry reservation, O(1) hint merge, and shared audit coordination. Status: Proposed. Passes scripts/adr-governance.py. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Size bootstrap capacity-rejection expiry from the full configured sync cycle, await aborted engine tasks before claiming shutdown quiescence, and apply paid-list edge flexibility against the runtime group width. Also fixes the all-target Clippy gate and adds regression coverage.\n\nSemVer: patch
Comment on lines
+560
to
564
| let keys = self | ||
| .blocking_tracker | ||
| .spawn_blocking(move || -> Result<Vec<XorName>> { | ||
| let rtxn = env | ||
| .read_txn() |
Comment on lines
+605
to
+609
| let value = self | ||
| .blocking_tracker | ||
| .spawn_blocking(move || -> Result<Option<Vec<u8>>> { | ||
| let rtxn = env | ||
| .read_txn() |
Comment on lines
+1031
to
+1033
| let config = ReplicationConfig::default(); | ||
| let batches = config.neighbor_sync_scope / config.neighbor_sync_peer_count; | ||
| let cycle_cadence = config.neighbor_sync_interval_max * u32::try_from(batches).unwrap(); |
Comment on lines
+1283
to
+1287
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); |
Comment on lines
+279
to
+283
| fn paid_hint_survives_duplicate_replica_rejection() { | ||
| // With churned local views, a sender may believe we are in the storage | ||
| // close group while our own view rejects the replica hint. If the same | ||
| // key also arrives as a paid hint, paid-list admission must still run. | ||
| let key = xor_name_from_byte(0xEE); |
mickvandijke
force-pushed
the
fix/neighbor-sync-drain-window
branch
from
8d73642 to
e0c0385
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR hardens replication repair under churn, load, queue pressure, event-stream failure, and shutdown. It expands the original neighbor-sync drain fix into a broader set of paid-list repair, verification, bootstrap accounting, audit concurrency, fresh-offer dispatch, admission gating, and async lifecycle fixes so legitimate repair work is not silently dropped, converted into false audit failures, delayed behind stale peers, left spinning on closed broadcasts, serialized behind a saturated worker pool and dropped through broadcast lag, used to conscript out-of-range nodes into storing arbitrary keys, downloaded and stored under responsibility decisions that topology churn had already invalidated, left holding LMDB/P2P resources — including detached LMDB blocking transactions — after engine shutdown, or wedged in permanent bootstrap by a peer-removal race that orphans capacity-rejection accounting.
The branch contains twenty-seven commits:
19bac10-fix(replication): harden paid-list verification repair0a5f078-fix(replication): tolerate paid-list edge churn2576f7a-fix(replication): drain priority sync queue and recover from routing-event lag95a8cdf-fix(replication): preserve verification retry capacity0c7bd9a-fix(replication): eliminate false audit challenge timeouts3800f9a-fix(replication): preserve dequeued retry reservations95fb3d9-chore(replication): fix audit admission clippy6ccb9bf-fix(replication): release cancelled async workb8d490d-fix(replication): silence no-logging audit label warnings77aa87c-fix(replication): unblock bootstrap when rejected peer leavesc896fa4-refactor(replication): prioritize source-aware hints5accdfb-refactor(replication): aggregate bootstrap hint batches1545551-fix(replication): aggregate verification per peer6b701c6-fix(replication): drain fresh offers through LMDB writesd2dc3f5-fix(replication): track detached audit workf81d28e-fix(replication): stop message handler when event streams close4859608-fix(replication): prune departed peers during DHT lag recovery50f6d7e-fix(replication): penalize rejected singleton replica hintsb251ba2-fix(replication): keep fresh offers off the serial message loopde8dcd7-fix(replication): gate replica downloads on storage responsibilitya0d63ae-refactor(replication): admit hints through a single relevance gate3d83004-perf(replication): merge duplicate hints without rebuilding the fetch heap52d19ef-fix(replication): drain detached LMDB blocking ops on shutdown31e5067-fix(replication): expire orphaned capacity-rejection records to unstall bootstrap3f4bbe0-fix(replication): recheck storage responsibility at the point of downloadce29af7-docs(adr): record replication repair hardening decisions (PR #165)8d73642-fix(replication): address deep review findingsProblems Addressed
Notifycoalescing allowed queued priority peers to wait for later 10-20 minute periodic ticks, while lagged DHT broadcasts could hide entrants entirely or leave departed peers ahead of current neighbors, each consuming a request timeout.spawn_blockingwaiter does not cancel the blocking transaction, so shutdown could report drained while LMDB still owned the environment.RecvError::Closedcould consume a core, flood P2P warnings, and prevent the replication pipeline from shutting down cleanly.key[0] % 64) collapsed the close-prefix accepted-key set onto a single shard, making the inline path the steady state, backing up the 256-slot inbound queue, and dropping replication messages wholesale once the P2P event loop lagged its broadcast receiver.HintPipeline::Replicaskipped theis_responsible(storage_admission_width)gate that paid keys had to pass, and a stored pipeline tag let a second replica message escalate a queued paid entry through thealready_pendingfast path, so a peer could name any key and force nodes ranked 10-20 for it to fetch and store it.select!on the shutdown token against futures awaiting aspawn_blockingLMDB transaction — the fetch worker'sstorage.put, the prune pass'sstorage.delete/paid_list.remove_batch, the verification worker'spaid_list.insert. Dropping the losing future does not cancel the blocking closure, which keeps running with a clonedEnv; per-fetch tasks were also bare-spawned outside both trackers, so a droppedin_flightset could leakArc<LmdbStorage>pastshutdown(). Reopening the same environment afterwards was undefined behavior.PeerRemovedcleanup — but the note sites and the removal handler run on different tokio tasks, with await points between the last "P is live" observation and the insert. A removal fully processed inside that window made its clear a no-op, and the subsequent note recorded an entry no future event could retire:check_bootstrap_drainedreturned false forever, audits stayed disabled (Invariant 19), and the node advertisedbootstrapping: trueindefinitely, drawing network-wide bootstrap-claim trust penalties — reachable deliberately by overflowingpending_verifyduring a victim's bootstrap and disconnecting. Adjacent liveness gap: every drain check was event-driven and a clean-cycle clear never re-checked, so a quiet node could satisfy the drain condition with nothing left to observe it.types.rsandadmission.rsthat responsibility is decided against live routing state at the point of download.Changes
Paid-list verification and repair
pending_verify, fetch queue, and in-flight fetch state.Paid-list edge churn
Neighbor sync and bootstrap lifecycle
HashMap<PeerId, Instant>instead of a bare set) and expires records only after a runtime-derived full-cycle window covering every neighbor batch, per-peer request deadlines, cooldown, and one slow-cadence interval of slack (125 minutes with defaults), forfeiting abandoned/departed-source debt consistently with peer-removal cleanup.pending_peer_requestsearly-returns: pending requests legitimately block the drain check itself but no longer block expiry, and a drain condition that became true without a triggering event is now observed within one tick.Source-aware bounded verification
Fresh-offer dispatch, admission gating, and fetch-queue merges
handle_fresh_offerhas a single caller and no inline verification/LMDB path on the serial loop.key[0] % 64shard locks with an exact per-key in-flight set behind an RAII guard, so unrelated keys never contend and concurrent duplicates collapse onto the first claimant instead of repeating its verification.replica_hint_sourcesinstead of storing the tag, deleting the paid→replica escalation and both demotion sites.is_responsible(storage_admission_width)at both fetch sites (local paid-list fast path and post-verification path), matching pruning width; fresh PUTs keep their wider accept window.paid_list_close_group_size(20), so mislabelling a hint gains nothing and the two-gate rescue dance (rejected_replicarescued byadmitted_paid) is removed; the admissible key set is unchanged.FetchCandidateintoFetchOrder(key + distance, the only fields theOrdimpl reads) held in the heap andFetchPayload(sources + retry metadata) held in a key-indexed map that also serves as the membership index, making a duplicate-source merge an O(1) map lookup and rebuilding the heap only when a departed peer actually orphans a candidate.is_responsible(storage_admission_width)on every fetch attempt insideexecute_single_fetch— before spending bandwidth (per-source retries re-enter there, so they are covered) and once more beforestorage.put, so bytes arriving after responsibility lapsed mid-round-trip are not written. A lapsed attempt resolves asFetchResult::NoLongerResponsible, which sharesStored's terminal path (retry-slot release plus bootstrap accounting) and reports no trust event; the verification-time check remains as a cheap pre-filter that keeps never-responsible keys out of the fetch queue.apply_fetch_result, so eachFetchResultvariant's queue transition is unit-testable without a live network.Audit concurrency and observability
AuditChallengeCoordinatoracross responsible, prune-confirmation, and possession audits.Event-stream and detached-task lifecycle
LmdbStorageandPaidListroute everyspawn_blockingthrough per-instanceTaskTrackers and exposewait_idle(), so a dropped async awaiter no longer untracks a live transaction (constructor-time opens stay untracked — they cannot outlive the constructor).shutdown()awaits timed-out long-lived tasks after requesting abort, then awaitswait_idle()on both environments after the detached-task drain; when it returns, producer tasks have actually exited, no LMDB blocking operation is still running, and no engine-spawned task holdsArc<LmdbStorage>/Arc<PaidList>.tokio::spawn; thein_flightplumbing and prompt network-I/O cancellation are unchanged — only the bounded in-flight LMDB transaction is awaited.Latest Commit Details
8d73642- address deep review findingsCloses four review findings before merge. Bootstrap capacity-rejection expiry is now derived from the runtime neighbor scope, batch width, slowest cadence, per-peer request deadline, cooldown, and one cadence interval of slack; with defaults the window is 125 minutes rather than 70, so a live source near the end of a five-batch cycle is not forfeited before its next legitimate re-hint.
shutdown()now awaits a long-lived task after requestingabort, ensuring synchronous sections have actually exited and dropped storage/P2P clones before detached work and LMDB trackers are declared quiescent. Paid-list flexible-edge activation now usesReplicationConfig::paid_list_close_group_size, preserving strict majority for an undersized 20/24 group and enabling the intended four-edge tolerance for a full non-default 16-peer group. The commit also fixes the all-target/all-feature Clippy gate and adds focused regression coverage. Validation: 738 all-feature library tests, 434 no-default-feature replication tests, the shutdown-LMDB PoC, formatting, diff checks, and the exact CI Clippy command all pass.ce29af7- record replication repair hardening decisionsAdds ADR-0005, documenting the repair-hardening decisions, invariants, operational trade-offs, and lifecycle guarantees implemented by this PR.
3f4bbe0- recheck storage responsibility at the point of downloadde8dcd7gated replica downloads onis_responsible(storage_admission_width), but only at verification-completion time — the answer was never re-asked before the chunk was actually downloaded and stored. The fetch queue is nearest-first and up to 131,072 entries deep, so a far candidate can wait unboundedly long while closer keys jump ahead, and the per-source retry path reused the same stale decision for every fallback source. Topology churn between promotion and download therefore still ended in a download, a disk write, and later fetch→store→prune churn once pruning evicted the record at the same width — whiletypes.rsandadmission.rsboth documented that responsibility is decided against live routing state at the point of download. Responsibility is now rechecked per fetch attempt insideexecute_single_fetch, where no queue lock is held: once at the top before spending bandwidth (retries re-enter there), and once more beforestorage.putso bytes that arrive after responsibility lapsed mid-round-trip are not written; edge flapping is dampened by the marginstorage_admission_widthalready adds overclose_group_size. A lapsed attempt resolves as the newFetchResult::NoLongerResponsible, which deliberately sharesStored's terminal path —complete_fetchreleases the verification retry-slot reservation and the worker's terminal handling shrinks the bootstrap pending set, so a declined key cannot stall bootstrap drain — and reports no trust event, since the source did nothing wrong. The promotion-time check stays as a cheap pre-filter; event-driven purging of the fetch queue on routing events was considered and rejected (O(queue) scans per churn event, racy, strictly more complex than the lazy recheck). The worker's result handling is extracted intoapply_fetch_resultfor direct unit coverage, and a new e2e drives a live 12-node network through a test-only seam that enqueues a fetch candidate for a key the target is not responsible for — the exact state a stale promotion leaves behind, since live topology cannot be shifted deterministically inside the millisecond promotion→dequeue window — asserting the chunk is never stored and the key exits the pipeline terminally, with an in-responsibility positive control through the same seam and holder. With the rechecks disabled the e2e fails by storing the stale chunk, confirming it discriminates.31e5067- expire orphaned capacity-rejection records to unstall bootstrap77aa87cmadePeerRemovedretire a departed source's capacity-rejection record, but the record's lifecycle was still purely event-driven and the note/removal paths run on different tokio tasks. Each note path has await points (LMDB reads, lock acquisitions) between its last "source is live" observation and the insert, so a removal fully processed inside that window cleared nothing and the subsequent note recorded an entry that no later admission cycle or removal event could ever retire — permanently blocking drain, disabling all audits (Invariant 19), and leaving the node claiming bootstrap network-wide, where other peers' bootstrap-claim abuse logic trust-penalizes it. An attacker could reach this deliberately by overflowing a victim'spending_verifyduring bootstrap and disconnecting immediately. The record now carries its most recent rejection time and expires after a runtime-derived full neighbor-cycle window (all configured batches and request deadlines, the cooldown floor, plus one slow-cadence interval of slack; 125 minutes with defaults), with expiry and a drain re-check run from the verification worker tick ahead of the pending-request early-returns. That same tick closes the adjacent liveness gap where a clean-cycle clear satisfied the drain condition with no event left to observe it. Expiry forfeits the keys the departed source still owed — consistent withupdate_bootstrap_after_peer_removed, which already forfeits a departed source's owed work wholesale; post-bootstrap neighbor sync and audit/repair recover them. Deliberately not a routing-table membership check at the note sites (only shrinks the TOCTOU window) and not a global bootstrap deadline (would change Invariant 19 semantics for genuinely busy bootstraps). Regression coverage reproduces the exact race ordering: removal cleanup first as a no-op, the rejection recorded after, drain asserted blocked, then TTL expiry drains it.b251ba2- keep fresh offers off the serial message loopOnce all four worker permits were held,
dispatch_fresh_offerhandled the next offer — an on-chain payment verification and a multi-MiB LMDB write — inline on the serial non-audit loop, backing up the 256-slot inbound queue until the P2P event loop also handled messages inline and its broadcast receiver lagged and dropped replication messages wholesale. The per-key shard locks (key[0] % 64) made that the steady state rather than the exception: a node only receives fresh offers for keys close to its own ID, so the accepted key set shares a long prefix and nearly every offer landed on one shard. The ordering those locks preserved has no meaning here, since the key is the content address of the data. Dispatch now only claims the key, takes an admission permit, and spawns; the worker permit is awaited inside the task sohandle_fresh_offerhas a single caller and no inline fallback exists. A 16-permit admission semaphore bounds outstanding offers at a 64 MiB ceiling, and an exact per-key in-flight set behind an RAII guard replaces the shard locks so unrelated keys never contend and concurrent duplicates collapse onto the first claimant. Past the bound an offer is refused — not stored, and therefore penalized as absent by the delayed possession check — rather than queued or handled inline.de8dcd7- gate replica downloads on storage responsibilityA replica hint is a possession claim, but the receiver treated it as authorization:
HintPipeline::Replicaskipped theis_responsible(storage_admission_width)check thatPaidOnlykeys had to pass, so whoever labelled the hint decided what we store. Two messages were enough to exploit it — a paid hint queued key K asPaidOnly, then a replica re-advertisement hit thealready_pendingfast path inadmit_hintsand escalated the live entry toReplica, and both fetch paths downloaded without ever asking whether this node was in the top-9 for K. The tag is now derived fromreplica_hint_sourcesinstead of stored (the tag is "did any peer claim to hold this"), which deletes the escalation and both demotion sites, and downloads are gated on live routing state at both diverging fetch sites. This narrows replica repair from the sender's view to our own — a transiently skewed routing table now declines to repair a key it ranks outside the top-9, matching pruning, which already evicts at the same width — while fresh PUTs keep their wider accept window.a0d63ae- admit hints through a single relevance gateAdmission previously asked two questions depending on the sender's label — replica hints gated at
storage_admission_width(9), paid hints atpaid_list_close_group_size(20) — letting the sender choose its own gate. Admission now decides only relevance (should we learn this key exists and is paid for) at the 20-wide paid close group, since that is the width across which nodes track payment validity; storage responsibility stays at download time against live routing state. Mislabelling a hint gains an attacker nothing because both labels reach the same gate; the admissible key set is unchanged (a paid hint for any key a replica hint could carry was already admissible at 20), and the two-gate rescue dance is removed. It does recover information the old gate discarded: a replica hint for a key we rank 10-20 for was previously rejected outright even though that band exists precisely to track payment validity.3d83004- merge duplicate hints without rebuilding the fetch heapA hint for a key already in the fetch queue only needs its advertiser merged into that candidate's source set — a field the heap never orders on — but the old path took the whole heap, scanned it linearly for the key, and re-heapified, costing O(n) per duplicate; under source aggregation a batch of m duplicates ran O(m·n) while holding the global queue write lock, which a neighbor could trigger by re-hinting queued keys.
FetchCandidateis split intoFetchOrder(key + distance, the only fields theOrdimpl reads) held in the heap andFetchPayload(sources + retry metadata) held in a key-indexed map that also serves as the membership index; merging a source is now an O(1) map lookup that never touches the heap, and the departed-peer path edits payloads in place and rebuilds only when a peer actually orphaned a candidate. Measured per-key merge cost goes from 302µs at a 50k-deep queue to a flat ~400ns independent of depth.52d19ef- drain detached LMDB blocking ops on shutdownReplicationEngine::shutdown()promises that when it returns no background work still holds the LMDB environment, but several paths race aselect!on the shutdown token against futures awaiting aspawn_blockingLMDB transaction — the fetch worker'sstorage.put, the neighbor-sync prune pass'sstorage.deleteandpaid_list.remove_batch(thePaidListis a second LMDB environment with the same pattern), and the verification worker'spaid_list.insert. Dropping the losing future does not cancelspawn_blocking: the closure keeps running on the blocking pool with a clonedEnv, so reopening the same environment after shutdown was undefined behavior. Rather than shielding every call site, the blocking tasks are now tracked at the storage layer:LmdbStorageandPaidListroute everyspawn_blockingthrough their ownTaskTrackerand exposewait_idle(), whichshutdown()awaits after its existing detached-task drain. Per-fetch tasks are additionally spawned on the detached tracker, so an aborted worker's droppedin_flightset can no longer leakArc<LmdbStorage>past shutdown. Cancellation semantics are unchanged — network I/O still aborts promptly, and shutdown waits only for the bounded in-flight transaction (milliseconds). Regression coverage parks a write inside its blocking closure, drops the awaiter, and proveswait_idle()/shutdown()block until it commits and both environments reopen cleanly.Preceding Follow-up Commit Details
f81d28e- stop the message handler when event streams closeMakes P2P lag remain recoverable but treats a closed P2P broadcast as terminal, matching the new terminal handling for a closed DHT broadcast. The replication loop now exits instead of spinning indefinitely on an immediately ready closed receiver; dropping its sender also lets the serial handler finish. The commit additionally hoists imports required by the Rust conventions hook.
4859608- prune departed peers during DHT lag recoveryBrings lag recovery in line with the normal
KClosestPeersChangedpath by retaining only the current close-peer set before requeueing its members. Stale peers from missed departure events no longer sit ahead of genuine entrants and burn one request timeout each.50f6d7e- penalize rejected singleton replica hintsPenalizes a sole replica advertiser when either the close group definitively rejects the key or that advertiser explicitly denies possessing it. This closes the free-replication abuse path while keeping inconclusive rounds without a direct contradiction, paid-only advertisements, and corroborated replica hints non-penalizing. Trust reports remain bounded per peer and verification cycle.
Earlier Follow-up Commit Details
77aa87c- unblock bootstrap when a rejected peer leavesRetires capacity-rejection debt on
PeerRemovedand immediately reruns bootstrap drain detection, so progress no longer depends on an unrelated later pipeline event.c896fa4- prioritise source-aware hintsReplaces the single hint sender with live source sets, preserves replica claimants separately, prioritises corroborated work, bounds verification cycles and network concurrency, and simplifies stale proof-of-concept tests around the production queue implementation.
5accdfb- aggregate bootstrap hint batchesMakes a bootstrap neighbor batch an atomic source-aggregation unit: sync requests run concurrently, response metadata is processed first, admitted hints are queued together, and verification waits until batch publication and drain accounting are complete.
1545551- aggregate verification per peerSends one bounded request per target peer for the cycle, aligns the incoming limit with the cycle bound, and rejects oversized batches rather than processing a misleading prefix.
6b701c6- drain fresh offers through LMDB writesRemoves cancellation around a started fresh-offer handler and makes its tracker drain unconditional, preventing a dropped async waiter from detaching a live blocking LMDB transaction.
d2dc3f5- track detached audit workGeneralises the fresh-offer tracker into a shared detached-task lifecycle barrier and applies it to storage/P2P-capable audit responders, delayed possession checks, and audit launches.
Coverage Added Or Updated
wait_idleregression tests: a write parked inside its blocking closure with a dropped awaiter keepswait_idleblocked, commits after release, and leaves the store usable.poc_shutdown_lmdb_drain):shutdown()blocks until a detached LMDB write commits, after which both environments reopen cleanly.NoLongerResponsible(terminal exit, retry-slot release, no verification requeue), terminal-path parity withStored, and preserved source-failure retry/requeue transitions.Expected Effect
shutdown()returns, timed-out long-lived tasks have been aborted and joined, no LMDB blocking operation is still running on either environment, and no engine-spawned task holds the storage, so the same LMDB files can be reopened safely.SemVer
Patch.